package cn.jly.bigdata.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author lanyangji
 * @date 2019/11/24 10:47
 */
object Spark02_expr2 {

  def main(args: Array[String]): Unit = {

    // local模式

    // 创建 SparkConf 对象
    // 这边也自定义了本地的模式（分配多少个cpu）
    // app id
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount")

    // 创建spark上下文对象，SparkContext -> sc
    val sc = new SparkContext(sparkConf)

    // mapPartitions算子
    val listRdd2: RDD[Int] = sc.makeRDD(1 to 10)
    // f: Iterator[T] => Iterator[U]
    //val mapPartRdd: RDD[Int] = listRdd2.mapPartitions(_.map(_ * 2))
    // mapPartitions可以对一个RDD的所有分区进行遍历，一个分区循环一遍，n个分区循环n遍
    // 效率优于map算子，减少了发送到执行器的交互次数
    val mapPartRdd: RDD[Int] = listRdd2.mapPartitions(datas => {
      println
      // 这个map 是scala中的map，不是RDD中的map
      datas.map(_ * 2)
    })
    mapPartRdd.collect.foreach(println)
  }
}
